use actix_web::{
    dev::{forward_ready, Service, ServiceRequest, ServiceResponse, Transform},
    Error, HttpMessage,
};
use futures_util::future::LocalBoxFuture;
use std::{
    collections::HashMap,
    future::{ready, Ready},
    rc::Rc,
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

#[derive(Clone)]
struct RateLimitEntry {
    count: u32,
    window_start: Instant,
}

#[derive(Clone)]
pub struct RateLimit {
    store: Arc<Mutex<HashMap<String, RateLimitEntry>>>,
    max_requests: u32,
    window_duration: Duration,
}

impl RateLimit {
    pub fn new(max_requests: u32, window_duration: Duration) -> Self {
        Self {
            store: Arc::new(Mutex::new(HashMap::new())),
            max_requests,
            window_duration,
        }
    }

    pub fn per_minute(max_requests: u32) -> Self {
        Self::new(max_requests, Duration::from_secs(60))
    }

    pub fn per_hour(max_requests: u32) -> Self {
        Self::new(max_requests, Duration::from_secs(3600))
    }

    fn is_allowed(&self, key: &str) -> bool {
        let mut store = self.store.lock().unwrap();
        let now = Instant::now();

        match store.get_mut(key) {
            Some(entry) => {
                // Check if window has expired
                if now.duration_since(entry.window_start) >= self.window_duration {
                    // Reset window
                    entry.count = 1;
                    entry.window_start = now;
                    true
                } else if entry.count < self.max_requests {
                    // Increment count
                    entry.count += 1;
                    true
                } else {
                    // Rate limit exceeded
                    false
                }
            }
            None => {
                // First request for this key
                store.insert(
                    key.to_string(),
                    RateLimitEntry {
                        count: 1,
                        window_start: now,
                    },
                );
                true
            }
        }
    }

    fn cleanup_expired(&self) {
        let mut store = self.store.lock().unwrap();
        let now = Instant::now();
        
        store.retain(|_, entry| {
            now.duration_since(entry.window_start) < self.window_duration
        });
    }
}

impl<S, B> Transform<S, ServiceRequest> for RateLimit
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type InitError = ();
    type Transform = RateLimitService<S>;
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(RateLimitService {
            service: Rc::new(service),
            rate_limit: self.clone(),
        }))
    }
}

pub struct RateLimitService<S> {
    service: Rc<S>,
    rate_limit: RateLimit,
}

impl<S, B> Service<ServiceRequest> for RateLimitService<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;

    forward_ready!(service);

    fn call(&self, req: ServiceRequest) -> Self::Future {
        let service = self.service.clone();
        let rate_limit = self.rate_limit.clone();

        Box::pin(async move {
            // Get client identifier (IP address)
            let client_ip = req
                .connection_info()
                .realip_remote_addr()
                .unwrap_or("unknown")
                .to_string();

            // Check rate limit
            if !rate_limit.is_allowed(&client_ip) {
                tracing::warn!("Rate limit exceeded for IP: {}", client_ip);
                return Err(actix_web::error::ErrorTooManyRequests(
                    "Rate limit exceeded"
                ));
            }

            // Periodically cleanup expired entries (simple approach)
            if rand::random::<f32>() < 0.01 {
                rate_limit.cleanup_expired();
            }

            service.call(req).await
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use actix_web::{test, web, App, HttpResponse};

    async fn test_handler() -> HttpResponse {
        HttpResponse::Ok().json(serde_json::json!({"message": "test"}))
    }

    #[actix_web::test]
    async fn test_rate_limit_middleware() {
        let rate_limit = RateLimit::new(2, Duration::from_secs(60));
        let app = test::init_service(
            App::new()
                .wrap(rate_limit)
                .route("/test", web::get().to(test_handler))
        ).await;

        // First request should succeed
        let req1 = test::TestRequest::get().uri("/test").to_request();
        let resp1 = test::call_service(&app, req1).await;
        assert!(resp1.status().is_success());

        // Second request should succeed
        let req2 = test::TestRequest::get().uri("/test").to_request();
        let resp2 = test::call_service(&app, req2).await;
        assert!(resp2.status().is_success());

        // Third request should be rate limited
        let req3 = test::TestRequest::get().uri("/test").to_request();
        let resp3 = test::call_service(&app, req3).await;
        assert_eq!(resp3.status(), 429);
    }
}